package com.sunday.common.mq.rocket.brave.study.apache.e4_Message_Tracing;

import com.sunday.common.mq.rocket.brave.study.apache.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.stream.Stream;

/**
 * 邮件跟踪
 * https://github.com/apache/rocketmq-spring/wiki/Message-Tracing
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class MessageTracingApi {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private static final String topic = "your_topic";

    @GetMapping("send")
    public void send() {
        Stream.iterate(0, i -> i < 1, i -> i + 1).forEach(i -> {
            User user = User.builder().id(i).userName("john" + i).userAge(33 + i).build();
            Message<User> message = MessageBuilder.withPayload(user).build();
            rocketMQTemplate.send(topic, message);
        });
    }

    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = "your_topic",
            consumerGroup = "your_consumer_group_name",
            consumeThreadNumber = 1,
            enableMsgTrace = true,
            customizedTraceTopic = "my-trace-topic"
    )
    public static class MessageExtConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
        @Override
        public void onMessage(MessageExt message) {
            log.info("{} - {}", new String(message.getBody()), message);
        }

        /**
         * 如何指定使用者从何处开始使用消息？
         * 默认消费偏移量请参考：RocketMQ常见问题解答。
         * 若要自定义使用者的起始位置，只需向使用者类添加一个接口实现。
         * 示例如下：RocketMQPushConsumerLifecycleListener
         */
        @Override
        public void prepareStart(DefaultMQPushConsumer consumer) {
            // 从现在开始设置消费者消费消息
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
            consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
        }
    }

}